package com.sunzm.flink.datastream.scala.window

import java.time.Duration

import com.alibaba.fastjson.{JSON, JSONObject}
import org.apache.commons.lang3.StringUtils
import org.apache.commons.lang3.time.{DateUtils, FastDateFormat}
import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, WatermarkStrategy}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector
import org.slf4j.{Logger, LoggerFactory}

/**
 *
 * 基于事件时间的滚动窗口
 *
 * @author Administrator
 * @version 1.0
 * @date 2021-06-22 20:50
 */
object ScalaEventTimeWindowDemo {
  private val logger: Logger = LoggerFactory.getLogger(this.getClass.getName.stripSuffix("$"))
  private val isLocal = true

  def main(args: Array[String]): Unit = {

    //1.创建执行的环境
    val env: StreamExecutionEnvironment = if (isLocal) {
      StreamExecutionEnvironment.createLocalEnvironmentWithWebUI()
    } else {
      StreamExecutionEnvironment.getExecutionEnvironment
    }

    //并行度设置成1，便于观察
    //不然，需要每个分区的watermark的时间都要进展到窗口关闭时间，窗口才会触发计算
    env.setParallelism(1)

    val dataStream: DataStream[String] = env.socketTextStream("82.156.210.70", 9999)

    //创建一个Watermark生成策略对象，并提取事件时间
    val watermarkStrategy = WatermarkStrategy.forBoundedOutOfOrderness[CallLog](Duration.ofSeconds(1))
      .withTimestampAssigner(new SerializableTimestampAssigner[CallLog] {
        override def extractTimestamp(element: CallLog, recordTimestamp: Long): Long = {

          DateUtils.parseDate(element.dateTimeStr, "yyyyMMddHHmmssSSS").getTime
        }
      })
      //只要任何一个分区Watermark到时间了，这个分区的数据就会触发执行，如果不设置这个，需要所有分区都收到新的消息才行
      //如果加了下面的配置，必须满足2个条件其中之一，才会触发：
      //(1)所有分区Watermark到时间了
      //(2)任意一个分区Watermark到时间了，但是这个分区超过10分钟没收到数据了
      .withIdleness(Duration.ofMinutes(10))

    //dataStream.windowAll()
    // 先转换成CallLog格式,再进行keyBy
    val keyedDS: KeyedStream[CallLog, String] = dataStream.flatMap((line, out: Collector[CallLog]) => {
      if (StringUtils.isNotBlank(line)) {

        try {
          val jSONObject: JSONObject = JSON.parseObject(line)

          val callId: String = jSONObject.getString("callId")
          val dateTimeStr: String = jSONObject.getString("dateTimeStr")

          out.collect(CallLog(callId, dateTimeStr))
        } catch {
          case e: Throwable => {
            logger.warn(s"收到格式异常数据: ${line}", e)
          }
        }
      }
    })
      //设置Watermark生成策略，并提取事件时间
      .assignTimestampsAndWatermarks(watermarkStrategy)
      .keyBy(_.callId)

    //10秒的滚动窗口,基于事件时间的
    //{"callId":"call-202106260001","dateTimeStr":"20210627165616000"}
    //{"callId":"call-202106260001","dateTimeStr":"20210627165612000"}
    //{"callId":"call-202106260002","dateTimeStr":"20210627165631001"}
    val resultDS: DataStream[String] = keyedDS
      //基于事件时间开一个10秒的窗口
      .window(TumblingEventTimeWindows.of(Time.seconds(10)))
      //允许数据迟到5秒，watermark到了窗口关闭时间，触发一次计算，但是不关闭窗口
      //只有在接下来的5秒来的数据，每来一条都会触发一次计算
      //这5秒过去了，窗口就会真正关闭
      //注意：这个参数只在事件时间窗口中有效
      .allowedLateness(Time.seconds(5))
      .process(new MyProcessWindowFunction)

    resultDS.print("结果数据")

    //5.执行
    env.execute(this.getClass.getSimpleName.stripSuffix("$"))
  }

  //封装数据的样例类
  private case class CallLog(callId: String, dateTimeStr: String){
    override def toString = callId + "," + dateTimeStr
  }

  private class MyProcessWindowFunction extends ProcessWindowFunction[CallLog, String, String, TimeWindow] {

    override def process(key: String,
                         ctx: Context,
                         elements: Iterable[CallLog],
                         out: Collector[String]): Unit = {

      logger.info("触发窗口操作....")

      val builder = new StringBuilder

      //${System.lineSeparator()表示换行
      builder.append(s"key: ${key} ").append(System.lineSeparator)

      //把这个窗口内的所有数据拼装到一起
      elements.foreach(callLog => {
        builder.append(callLog.toString).append(System.lineSeparator)
      })

      //获取窗口开始时间
      val windowStart: Long = ctx.window.getStart
      //获取窗口结束时间
      val windowEnd: Long = ctx.window.getEnd
      //获取Watermark
      val currentWatermark: Long = ctx.currentWatermark

      val fastDateFormat = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss:SSS")

      val windowStartStr = fastDateFormat.format(windowStart)
      val windowEndStr = fastDateFormat.format(windowEnd)
      val currentWatermarkStr = fastDateFormat.format(currentWatermark)


      out.collect(s"Watermark: ${currentWatermarkStr}, 窗口范围：[${windowStartStr}, ${windowEndStr}], ${builder}")
    }
  }
}
